// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: hyuan

#include "coin2/exchange/kucoin/order_v1/parser.h"

#include <memory>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/kucoin/api_order/native_private_client.h"
#include "coin2/exchange/kucoin/order_v1/error_code.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::kucoin::order_v1 {

namespace {

using namespace coin2::exchange::kucoin::order_v1::error_code;

using coin::proto::AccountBalance;
using coin::proto::AccountPosition;
using coin::proto::CancelOrderResponse;
using coin::proto::CurrencyTransferElement;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;
using coin::proto::TransferRequestProto;
using coin2::exchange::base::order::IOrderContext;
using coin2::exchange::kucoin::symbology::KucoinCurrency;
using coin2::exchange::kucoin::symbology::KucoinProduct;
using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;

inline int64_t GetProcOrderId(const JsonValue& data) {
  if (data.HasMember("clientOid") && data["clientOid"].IsString()) {
    if (!data["clientOid"].IsNull()) {
      return stoll(data["clientOid"].GetString());
    }
  }
  return 0;
}

inline coin::proto::OrderSide GetOrderSide(const std::string& side) {
  if (side == "buy") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "sell") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline coin::proto::OrderType GetOrderType(const std::string& type) {
  if (type == "limit") {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else if (type == "market") {
    return coin::proto::OrderType::MARKET_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

inline void GetOrderStatus(ProductOrderElement* order, bool is_active, bool cancel_exist) {
  if (is_active) {
    order->set_is_live(true);
  } else {
    order->set_is_live(false);
    if (cancel_exist) {
      order->set_cancel_confirmed(true);
    } else {
      order->set_fully_filled(true);
    }
  }
}

inline void
SetOrderStatus(ProductOrderElement* order, const std::string& status, const std::string& type) {
  if ("open" == status || "match" == status) {
    // type:
    // - open
    // - match
    // - update
    order->set_is_live(true);
  } else if ("done" == status) {
    // type:
    // - cancelled
    // - filled
    order->set_is_live(false);
    if ("canceled" == type) {
      order->set_cancel_confirmed(true);
    } else if ("filled" == type) {
      order->set_fully_filled(true);
    }
  }
}

inline int64_t GetTimestamp(const std::string& time_str) { return stoll(time_str) * 1e6; }

inline bool IsKucoinSpotErrMessage(const JsonValue& doc) {
  if (doc.HasMember("code")) {
    return (std::string(doc["code"].GetString()) != "200000");
  }

  return false;
}

inline bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (json.HasMember("type") && json.HasMember("topic") && json.HasMember("data")) {
    return (json["type"] == "message") && (json["topic"] == "/spotMarket/tradeOrders");
  }
  return false;
}

inline bool IsBalanceUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (json.HasMember("type") && json.HasMember("topic") && json.HasMember("data")) {
    return (json["type"] == "message") && (json["topic"] == "/account/balance");
  }
  return false;
}

}  // namespace

void KucoinParser::ParseRestAccountCrossBalance() {
  /*
  [
    {
      "id": "5bd6e9286d99522a52e458de",
      "currency": "BTC",
      "type": "main",
      "balance": "237582.04299",
      "available": "237582.032",
      "holds": "0.01099"
    }
  ]
  */
  JsonDocument doc;
  if (!PrepareJsonDocument(&doc)) {
    LOG(ERROR) << "[ParseRestAccountBalance] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (doc.IsObject() && IsKucoinSpotErrMessage(doc)) {
    LOG(ERROR) << "[ParseCrossBalance] Error message\n" << JsonToString(doc);
    return;
  }

  if (!doc.HasMember("data") || doc["data"].IsNull() || !doc["data"].IsArray()) {
    LOG(ERROR) << "[ParseCrossBalance] Invalid data format\n" << JsonToString(doc);
    return;
  }

  for (const auto& data : doc["data"].GetArray()) {
    if (std::string_view(data["type"].GetString()) != "trade") {
      continue;
    }
    if (!data.HasMember("currency") || !data.HasMember("balance") || !data.HasMember("available") ||
        !data.HasMember("holds")) {
      continue;
    }
    CurrencyBalance balance;
    balance.set_currency_native(data["currency"].GetString());
    balance.set_total(stod(data["balance"].GetString()));
    balance.set_available(stod(data["available"].GetString()));
    balance.set_hold(stod(data["holds"].GetString()));
    raw_cross_balance_cache_map_[balance.currency_native()] = balance;
  }
}

void KucoinParser::ParseRestAccountBalance() {
  /*
  [
    {
      "id": "5bd6e9286d99522a52e458de",
      "currency": "BTC",
      "type": "main",
      "balance": "237582.04299",
      "available": "237582.032",
      "holds": "0.01099"
    }
  ]
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestAccountBalance] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_balance = ParseBalance(document);
  if (!account_balance.has_exchange() || !account_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountBalance KucoinParser::ParseBalance(const JsonValue& doc) {
  AccountBalance acc_balance;

  if (doc.IsObject() && IsKucoinSpotErrMessage(doc)) {
    LOG(ERROR) << "[ParseBalance] Error message\n" << JsonToString(doc);
    return acc_balance;
  }

  if (!doc.HasMember("data") || doc["data"].IsNull() || !doc["data"].IsArray()) {
    LOG(ERROR) << "[ParseBalance] Invalid data format\n" << JsonToString(doc);
    return acc_balance;
  }

  RawBalanceMap raw_balance_map;
  for (const auto& data : doc["data"].GetArray()) {
    if (std::string_view(data["type"].GetString()) != "trade_hf") {
      continue;
    }
    if (!data.HasMember("currency") || !data.HasMember("balance") || !data.HasMember("available") ||
        !data.HasMember("holds")) {
      continue;
    }
    CurrencyBalance balance;
    balance.set_currency_native(data["currency"].GetString());
    balance.set_total(stod(data["balance"].GetString()));
    balance.set_available(stod(data["available"].GetString()));
    balance.set_hold(stod(data["holds"].GetString()));
    raw_balance_map.emplace(balance.currency_native(), balance);
  }
  acc_balance.set_exchange("Kucoin");
  acc_balance.set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    auto* balance = acc_balance.add_each_balance();
    if (raw_balance_map.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(raw_balance_map.at(native_currency));
    }
    if (raw_cross_balance_cache_map_.count(native_currency)) {
      const auto& cross_bal = raw_cross_balance_cache_map_[native_currency];
      auto total = balance->total() + cross_bal.total();
      auto avail = balance->available() + cross_bal.available();
      auto hold = balance->hold() + cross_bal.hold();
      balance->set_total(total);
      balance->set_available(avail);
      balance->set_hold(hold);
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }

  is_account_balance_initialized_ = true;

  return acc_balance;
}

void KucoinParser::ParseRestOpenOrderInfoList() {
  /*
  {
    "currentPage": 1,
    "pageSize": 1,
    "totalNum": 153408,
    "totalPage": 153408,
    "items": [
      {
        "id": "5c35c02703aa673ceec2a168",
        "symbol": "BTC-USDT",
        "opType": "DEAL",
        "type": "limit",
        "side": "buy",
        "price": "10",
        "size": "2",
        "funds": "0",
        "dealFunds": "0.166",
        "dealSize": "2",
        "fee": "0",
        "feeCurrency": "USDT",
        "stp": "",
        "stop": "",
        "stopTriggered": false,
        "stopPrice": "0",
        "timeInForce": "GTC",
        "postOnly": false,
        "hidden": false,
        "iceberg": false,
        "visibleSize": "0",
        "cancelAfter": 0,
        "channel": "IOS",
        "clientOid": "",
        "remark": "",
        "tags": "",
        "isActive": false,
        "cancelExist": false,
        "createdAt": 1547026471000,
        "tradeType": "TRADE"
      }
    ]
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestOpenOrderInfoList] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (document.IsObject() && IsKucoinSpotErrMessage(document)) {
    LOG(ERROR) << "[ParseRestOpenOrderInfoList] Error message\n" << JsonToString(document);
    return;
  }
  if (!document.HasMember("data")) {
    LOG(ERROR) << "[ParseRestOpenOrderInfoList] Error message\n" << JsonToString(document);
    return;
  }
  if (document["data"].IsNull()) {
    LOG(WARNING) << "[ParseRestOpenOrderInfoList] empty open order list\n"
                 << JsonToString(document);
    return;
  }
  if (!document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRestOpenOrderInfoList] Invalid data format\n" << JsonToString(document);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  bool product_order_bundle_found = false;
  for (const auto& data : document["data"].GetArray()) {
    if (!data.HasMember("symbol")) {
      continue;
    }
    const std::string native_symbol = data["symbol"].GetString();
    if (!symcache_->has_product_native(native_symbol)) {
      continue;
    }
    auto order = product_order_bundle->add_each_order();
    ParseRestOrder(order, data);
    product_order_bundle_found = true;
  }

  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

void KucoinParser::ParseRestOrder(ProductOrderElement* order, const JsonValue& data) {
  order->set_exchange_type(ExchangeType::Kucoin);
  order->set_market_type(MarketType::Spot);
  order->set_api_version("v1");
  order->set_exchange_order_id(data["id"].GetString());
  order->set_symbol(data["symbol"].GetString());
  order->set_price(stod(data["price"].GetString()));
  order->set_qty(stod(data["size"].GetString()));
  order->set_proc_order_id(GetProcOrderId(data));
  order->set_side(GetOrderSide(data["side"].GetString()));
  order->set_order_type(GetOrderType(data["type"].GetString()));
  order->set_timestamp(msg_->timestamp);
  GetOrderStatus(order, data["active"].GetBool(), data["cancelExist"].GetBool());
}

void KucoinParser::ParseRestHistoryOrderInfoList() {
  /*
  {
    "currentPage": 1,
    "pageSize": 1,
    "totalNum": 153408,
    "totalPage": 153408,
    "items": [
      {
        "id": "5c35c02703aa673ceec2a168",
        "symbol": "BTC-USDT",
        "opType": "DEAL",
        "type": "limit",
        "side": "buy",
        "price": "10",
        "size": "2",
        "funds": "0",
        "dealFunds": "0.166",
        "dealSize": "2",
        "fee": "0",
        "feeCurrency": "USDT",
        "stp": "",
        "stop": "",
        "stopTriggered": false,
        "stopPrice": "0",
        "timeInForce": "GTC",
        "postOnly": false,
        "hidden": false,
        "iceberg": false,
        "visibleSize": "0",
        "cancelAfter": 0,
        "channel": "IOS",
        "clientOid": "",
        "remark": "",
        "tags": "",
        "isActive": false,
        "cancelExist": false,
        "createdAt": 1547026471000,
        "tradeType": "TRADE"
      }
    ]
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestHistoryOrderInfoList] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsKucoinSpotErrMessage(document)) {
    LOG(ERROR) << "[ParseRestHistoryOrderInfoList] Error message\n" << JsonToString(document);
    return;
  }

  if (!document.HasMember("data") || document["data"].IsNull() ||
      !document["data"].HasMember("items") || !document["data"]["items"].IsArray()) {
    LOG(ERROR) << "[ParseRestHistoryOrderInfoList] Invalid data format\n" << JsonToString(document);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  bool product_order_bundle_found = false;
  for (const auto& data : document["data"]["items"].GetArray()) {
    if (!data.HasMember("symbol")) {
      continue;
    }
    const std::string native_symbol = data["symbol"].GetString();
    if (!symcache_->has_product_native(native_symbol)) {
      continue;
    }
    auto order = product_order_bundle->add_each_order();
    ParseRestOrder(order, data);
    product_order_bundle_found = true;
  }

  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

void KucoinParser::ParseRestFillOrderInfoList() {
  /*
  {
    "currentPage":1,
    "pageSize":1,
    "totalNum":251915,
    "totalPage":251915,
    "items":[
      {
        "symbol":"BTC-USDT",
        "tradeId":"5c35c02709e4f67d5266954e",
        "orderId":"5c35c02703aa673ceec2a168",
        "counterOrderId":"5c1ab46003aa676e487fa8e3",
        "side":"buy",
        "liquidity":"taker",
        "forceTaker":true,
        "price":"0.083",
        "size":"0.8424304",
        "funds":"0.0699217232",
        "fee":"0",
        "feeRate":"0",
        "feeCurrency":"USDT",
        "stop":"",
        "type":"limit",
        "createdAt":1547026472000,
        "tradeType": "TRADE"
      }
    ]
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestFillOrderInfoList] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsKucoinSpotErrMessage(document)) {
    LOG(ERROR) << "[ParseRestFillOrderInfoList] Error message\n" << JsonToString(document);
    return;
  }

  if (!document.HasMember("data") || document["data"].IsNull() ||
      !document["data"].HasMember("items") || !document["data"]["items"].IsArray()) {
    LOG(ERROR) << "[ParseRestFillOrderInfoList] Invalid data format\n" << JsonToString(document);
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();
  bool product_fill_bundle_found = false;
  for (const auto& data : document["data"]["items"].GetArray()) {
    if (!data.HasMember("symbol")) {
      continue;
    }
    const std::string native_symbol = data["symbol"].GetString();
    if (!symcache_->has_product_native(native_symbol)) {
      continue;
    }
    auto fill = product_fill_bundle->add_each_fill();
    fill->Clear();
    ParseRestFillInfo(fill, data);
    product_fill_bundle_found = true;
  }

  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

void KucoinParser::ParseRestFillInfo(ProductFillElement* fill, const JsonValue& data) {
  fill->set_symbol(data["symbol"].GetString());
  fill->set_exchange_order_id(data["orderId"].GetString());
  fill->set_price(stod(data["price"].GetString()));
  fill->set_qty(stod(data["size"].GetString()));
  fill->set_fill_id(data["tradeId"].GetString());
  if (data.HasMember("fee") && !data["fee"].IsNull() && data.HasMember("feeCurrency") &&
      !data["feeCurrency"].IsNull()) {
    fill->set_fee(stod(data["fee"].GetString()));
    fill->set_fee_currency(data["feeCurrency"].GetString());
  }
  std::string liquidity = data["liquidity"].GetString();
  if (liquidity == "maker") {
    fill->set_fill_type(FillType::MAKER_FILL_TYPE);
  } else if (liquidity == "taker") {
    fill->set_fill_type(FillType::TAKER_FILL_TYPE);
  } else {
    fill->set_fill_type(FillType::UNKNOWN_FILL_TYPE);
  }
  fill->set_fill_timestamp(msg_->timestamp);
}

void KucoinParser::ParseRestBullet() {
  /*
  {
    "code": "200000",
    "data": {
      "instanceServers": [
        {
          "endpoint": "wss://push1-v2.kucoin.com/endpoint",
          "protocol": "websocket",
          "encrypt": true,
          "pingInterval": 50000,
          "pingTimeout": 10000
        }
      ],
      "token": "..."
    }
  }
  */
  if (is_bullet_initialized_) {
    return;
  }

  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestBullet] Parsing error.\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (IsKucoinSpotErrMessage(document)) {
    LOG(ERROR) << "[ParseRestBullet] Error message.\n" << JsonToString(document);
    return;
  }

  if (!document.HasMember("data") || !document["data"].HasMember("token") ||
      !document["data"].HasMember("instanceServers") ||
      !document["data"]["instanceServers"].IsArray() ||
      (document["data"]["instanceServers"].GetArray().Size() <= 0)) {
    LOG(ERROR) << "[ParseRestBullet] Invalid data format.\n" << JsonToString(document);
    return;
  }

  const auto& data = document["data"]["instanceServers"].GetArray()[0];
  if (!data.HasMember("endpoint") || !data.HasMember("encrypt") ||
      !data.HasMember("pingInterval") || !data.HasMember("pingTimeout")) {
    LOG(ERROR) << "[ParseRestBullet] Invalid bullet format.\n" << JsonToString(document);
    return;
  }

  std::vector<std::string> bullet_info;
  bullet_info.push_back(data["endpoint"].GetString());
  bullet_info.push_back(data["encrypt"].GetBool() ? "true" : "false");
  bullet_info.push_back(data["pingInterval"].GetString());
  bullet_info.push_back(data["pingTimeout"].GetString());
  bullet_info.push_back(document["data"]["token"].GetString());
  mb_->PublishEvent("kucoin/bullet", &bullet_info);
  is_bullet_initialized_ = true;
}

void KucoinParser::ParseSubmitResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_timestamp(GetCurrentTimestamp());

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[ParseSubmitResponse] AsioHttpContextStatus error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[ParseSubmitResponse] Parsing error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (json.HasMember("code") && json.HasMember("msg")) {
    auto error_code = GetOrderErrorCode(json["code"], json["msg"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[ParseSubmitResponse] Error response\n" << http_context->res;
      if (error_code == coin::proto::OrderErrorCode::ORDER_UNCERTAIN) {
        result_.Clear();
        result_.set_type(MsgType::IGNORABLE);
      } else {
        resp->set_success(false);
      }
      return;
    }
  }

  if (!json.HasMember("data") || !json["data"].HasMember("orderId")) {
    LOG(ERROR) << "[ParseSubmitResponse] Invalid data format\n" << JsonToString(json);
    resp->set_success(false);
    return;
  }
  resp->set_exchange_order_id(json["data"]["orderId"].GetString());

  resp->set_success(true);
}

void KucoinParser::ParseCancelResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[ParseCancelResponse] AsioHttpContextStatus error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[ParseCancelResponse] Parsing error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (json.HasMember("code") && json.HasMember("msg")) {
    auto error_code = GetOrderErrorCode(json["code"], json["msg"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[ParseCancelResponse] Error response. " << order_context->proc_order_id()
                 << "\n"
                 << http_context->res;
      if (error_code == coin::proto::OrderErrorCode::ORDER_UNCERTAIN) {
        result_.Clear();
        result_.set_type(MsgType::IGNORABLE);
      } else {
        resp->set_success(false);
      }
      return;
    }
  }

  if (json.HasMember("data") && json["data"].HasMember("orderId")) {
    resp->set_exchange_order_id(json["data"]["orderId"].GetString());
    resp->set_success(true);
    resp->set_cancel_confirmed(true);
  } else {
    LOG(ERROR) << "[ParseCancelResponse] Invalid data format\n" << JsonToString(json);
    resp->set_success(false);
  }
}

int32_t KucoinParser::ParseErrorMessage(const std::shared_ptr<AsioHttpContext>& http_context) {
  enable_time_.success = false;

  const auto http_code = http_context->res.result_int();
  if (http_code == 400) {
    LOG(ERROR) << "[ParseErrorMessage] Bad Request\n" << msg_->DebugString();
  } else if (http_code == 401) {
    LOG(ERROR) << "[ParseErrorMessage] Unauthorized\n" << msg_->DebugString();
  } else if (http_code == 403) {
    LOG(ERROR) << "[ParseErrorMessage] Forbidden or Too Many Requests\n" << msg_->DebugString();
  } else if (http_code == 404) {
    LOG(ERROR) << "[ParseErrorMessage] Not Found\n" << msg_->DebugString();
  } else if (http_code == 405) {
    LOG(ERROR) << "[ParseErrorMessage] Method Not Allowed\n" << msg_->DebugString();
  } else if (http_code == 415) {
    LOG(ERROR) << "[ParseErrorMessage] Unsupported Media Type\n" << msg_->DebugString();
  } else if (http_code == 429) {
    std::string body(http_context->res.body().data());
    if (body.find("1015") != std::string::npos) {
      enable_time_.enable_time = 30 * 1'000'000'000LL + GetCurrentTimestamp();
      enable_time_.success = true;
    } else if (body.find("200002") != std::string::npos) {
      enable_time_.enable_time = 10 * 1'000'000'000LL + GetCurrentTimestamp();
      enable_time_.success = true;
    }
    LOG(ERROR) << "[ParseErrorMessage] Rate Limit Exceeded\n" << msg_->DebugString();
  } else if (http_code == 500) {
    LOG(ERROR) << "[ParseErrorMessage] Internal Server Error\n" << msg_->DebugString();
  } else if (http_code == 503) {
    LOG(ERROR) << "[ParseErrorMessage] Service Unavailable\n" << msg_->DebugString();
  }

  return http_code;
}

void KucoinParser::ParseWsOrderInfoList(const JsonDocument& document) {
  /*
  {
    "type":"message",
    "topic":"/spotMarket/tradeOrders",
    "subject":"orderChange",
    "channelType":"private",
    "data":{
      "symbol":"KCS-USDT",
      "orderType":"limit",
      "side":"buy",
      "orderId":"5efab07953bdea00089965d2",
      "type":"open",
      "orderTime":1593487481683297666,
      "size":"0.1",
      "filledSize":"0",
      "price":"0.937",
      "clientOid":"1593487481000906",
      "remainSize":"0.1",
      "status":"open",
      "ts":1593487481683297666
    }
  }
  */
  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  const auto& data = document["data"];
  auto order = product_order_bundle->add_each_order();
  order->set_exchange_type(ExchangeType::Kucoin);
  order->set_market_type(MarketType::Spot);
  order->set_api_version("v1");
  order->set_exchange_order_id(data["orderId"].GetString());
  order->set_symbol(data["symbol"].GetString());
  order->set_price(stod(data["price"].GetString()));
  order->set_qty(stod(data["size"].GetString()));
  order->set_proc_order_id(GetProcOrderId(data));
  order->set_side(GetOrderSide(data["side"].GetString()));
  order->set_order_type(GetOrderType(data["orderType"].GetString()));
  order->set_timestamp(msg_->timestamp);
  SetOrderStatus(order, data["status"].GetString(), data["type"].GetString());

  result_.set_type(MsgType::ORDER_STATUS);
}

void KucoinParser::ParseWsTradeInfoList(const JsonDocument& document) {
  /*
  {
    "type": "message",
    "topic": "/spotMarket/tradeOrders",
    "userId": "62a7ece411e2d60001eee3c6",
    "channelType": "private",
    "subject": "orderChange",
    "data": {
      "symbol": "BTC-USDT",
      "orderType": "limit",
      "side": "sell",
      "orderId": "62b04633dbae9b00012d94ce",
      "liquidity": "maker",
      "type": "match",
      "orderTime": 1655719475474793000,
      "size": "0.00001",
      "filledSize": "0.00001",
      "price": "20630.1",
      "matchPrice": "20630.1",
      "matchSize": "0.00001",
      "tradeId": "62b046352e113d2923babb5a",
      "clientOid": "1655719428644838021",
      "remainSize": "0",
      "status": "open",
      "ts": 1655719477401642000
    }
  }
  */
  const auto& data = document["data"];
  const std::string type = data["type"].GetString();
  if (type != "match") {
    return;
  }
  const std::string symbol = data["symbol"].GetString();
  if (!symcache_->has_product_native(symbol)) {
    return;
  }
  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();
  auto fill = product_fill_bundle->add_each_fill();
  fill->set_symbol(symbol);
  fill->set_exchange_order_id(data["orderId"].GetString());
  fill->set_price(stod(data["matchPrice"].GetString()));
  fill->set_qty(stod(data["matchSize"].GetString()));
  fill->set_fill_id(data["tradeId"].GetString());
  const std::string liquidity = data["liquidity"].GetString();
  if (liquidity == "maker") {
    fill->set_fill_type(FillType::MAKER_FILL_TYPE);
  } else if (liquidity == "taker") {
    fill->set_fill_type(FillType::TAKER_FILL_TYPE);
  } else {
    fill->set_fill_type(FillType::UNKNOWN_FILL_TYPE);
  }
  fill->set_fill_timestamp(msg_->timestamp);

  result_.set_type(MsgType::ORDER_FILL);
}

void KucoinParser::ParseWsBalanceInfoList(const JsonDocument& document) {
  /*
  {
    "type": "message",
    "topic": "/account/balance",
    "subject": "account.balance",
    "channelType":"private",
      "data": {
        "total": "88",
        "available": "88",
        "availableChange": "88",
        "currency": "KCS",
        "hold": "0",
        "holdChange": "0",
        "relationEvent": "trade.setted",
        "relationEventId": "5c21e80303aa677bd09d7dff",
        "relationContext": {
          "symbol":"BTC-USDT",
          "tradeId":"5e6a5dca9e16882a7d83b7a4",
          "orderId":"5ea10479415e2f0009949d54"
        },
      "time": "1545743136994"
    }
  }
  */
  if (!is_account_balance_initialized_) {
    return;
  }
  auto acc_balance = result_.mutable_account_balance();
  acc_balance->Clear();
  const auto& data = document["data"];
  acc_balance->set_exchange("Kucoin");
  acc_balance->set_market_type("Spot");
  if (!symcache_->has_currency_native(data["currency"].GetString())) {
    return;
  }
  auto* currency = symcache_->GetCurrencyByNative(data["currency"].GetString());
  auto* currency_balance = acc_balance->add_each_balance();
  currency_balance->set_currency(currency->currency());
  currency_balance->set_currency_native(currency->native_currency());
  auto total = stod(data["total"].GetString());
  auto avail = stod(data["available"].GetString());
  currency_balance->set_total(total);
  currency_balance->set_available(avail);
  currency_balance->set_hold(total - avail);

  result_.set_type(MsgType::ACCOUNT_BALANCE);
}

void KucoinParser::ParseRestTransfers() {
  /*
{
  "code": "200000",
  "data": {
    "currentPage": 1,
    "pageSize": 50,
    "totalNum": 153,
    "totalPage": 4,
    "items": [
      {
        "id": "6322cc0d196d0f0001275d5f",
        "currency": "USDT",
        "amount": "100",
        "fee": "0",
        "balance": "0",
        "accountType": "MAIN",
        "bizType": "Sub-account transfer",
        "direction": "out",
        "createdAt": 1663224845903,
        "context": ""
      }
    ]
  }
}
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestTransfers] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (!document.HasMember("data") || document["data"].IsNull() || !document["data"].IsArray()) {
    result_.set_error_code(OrderErrorCode::INVALID_MSG_TYPE);
    LOG(ERROR) << "[ParseRestTransfers] Invalid data format\n" << JsonToString(document);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool product_order_bundle_found = false;
  for (const auto& data : document["data"].GetArray()) {
    std::vector<std::string> fields =
        {"currency", "amount", "id", "direction", "createdAt", "bizType"};
    if (!JsonFieldCheck(data, fields)) {
      LOG(ERROR) << "[ParseRestTransfers] Invalid transfer format" << JsonToString(data);
      continue;
    }

    std::string transfer_id = data["id"].GetString();
    std::string biz_type = data["bizType"].GetString();
    if (!transfer_first_record_) {
      if (transfer_map_[transfer_id] == "done") {
        continue;
      }
    }
    transfer_map_.insert_or_assign(transfer_id, "done");
    auto proto_type = TransferRequestProto::DEPOSIT;
    if (biz_type == "Deposit") {
      proto_type = TransferRequestProto::DEPOSIT;  // noted !!! WITHDRAW and DEPOSIT Waiting for
                                                   // validation
    } else if (biz_type == "Withdrawal") {
      proto_type = TransferRequestProto::WITHDRAW;
    } else if (biz_type == "Sub-account transfer") {
      proto_type = data["direction"] == "in" ? TransferRequestProto::DEPOSIT_SUBACCOUNT
                                             : TransferRequestProto::WITHDRAW_SUBACCOUNT;
    } else {
      continue;
    }

    int64_t create_timestamp = stoll(data["createdAt"].GetString()) * 1e6;
    auto currency =
        KucoinCurrency::FromNativeStrImpl(data["currency"].GetString(), create_timestamp);
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    currency_transfer->set_type(proto_type);
    currency_transfer->set_create_timestamp(create_timestamp);
    currency_transfer->set_currency(currency->currency());
    currency_transfer->set_qty(stod(data["amount"].GetString()));
    currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    currency_transfer->set_external_transfer_id(transfer_id);
    currency_transfer->set_from_address("");
    currency_transfer->set_to_address("");
    product_order_bundle_found = true;
  }

  if (!transfer_first_record_) {
    if (!is_transfer_deposit_initialized_) {
      is_transfer_deposit_initialized_ = true;
      LOG(INFO) << "[ParseRestDeposit] Deposit ignored: " << transfer_map_.size();
      return;
    }
  }
  if (product_order_bundle_found) {
    result_.set_type(MsgType::TRANSFER);
  }
}

void KucoinParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  DLOG(INFO) << "OG ws msg: " << JsonToString(document);
  if (IsOrderUpdate(document)) {
    ParseWsOrderInfoList(document);
    ParseWsTradeInfoList(document);
  } else if (IsBalanceUpdate(document)) {
    ParseWsBalanceInfoList(document);
  }
}

auto KucoinParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg_->type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  CHECK(msg.type == OrderSubTopicId::REST_MESSAGE);

  int32_t error_code = ParseErrorMessage(msg.http_context);
  if (error_code == 429) {
    if (enable_time_.success) {
      mb_->PublishEvent("kucoin/enable_time", reinterpret_cast<void*>(enable_time_.enable_time));
    }
  }

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::REST_ACCOUNT_CROSS_BALANCE:
      ParseRestAccountCrossBalance();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestAccountBalance();
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
      ParseRestOpenOrderInfoList();
      break;
    case OrderSubTopicId::REST_DONE_ORDERS:
      ParseRestHistoryOrderInfoList();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillOrderInfoList();
      break;
    // case KucoinOrderSubTopicId::REST_BULLET:
    case OrderSubTopicId::LISTEN_KEY:
      ParseRestBullet();
      break;
    case OrderSubTopicId::TRANSFER:
      ParseRestTransfers();
      break;
    default: {
      LOG(ERROR) << "[ParseRawMessage] Unknown query type\n" << msg_->DebugString();
      result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      break;
    }
  }

  return result_;
}

}  // namespace coin2::exchange::kucoin::order_v1
